package com.niodata.dp.plugin.core;

import com.google.common.base.Strings;
import com.niodata.dt.fs.DpRelativeLocalFileSystem;
import com.niodata.dt.fs.RelLocalFileSystemBuilder;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import javax.sql.DataSource;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.springframework.core.io.DefaultResourceLoader;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;


public class DefaultLocalPluginContext extends DefaultPluginContext implements PluginContext {

  Configuration hdfsConfiguration;
  private FileSystem fileSystem;
  private FileSystem tmpFileSystem;
  private DataSource dataSource;
  private File workDir;

  public DefaultLocalPluginContext() {
    super(null);
    this.workDir = new File(System.getProperty("user.dir"));
  }

  /**
   * init config.
   *
   * @param confMap config
   */
  @Override
  public void init(Map<String, Object> confMap) throws IOException {
    super.init(confMap);
    Configuration config = new Configuration();
    Set<Entry<String, Object>> entries = confMap.entrySet();
    Iterator<Entry<String, Object>> iterator = entries.iterator();
    while (iterator.hasNext()) {
      Entry<String, Object> entry = iterator.next();
      String key = entry.getKey() == null ? null : entry.getKey().toString();
      String value = entry.getValue() == null ? null : entry.getValue().toString();
      if (!Strings.isNullOrEmpty(key)) {
        config.set(key, value);
      }
    }
    String fsDir = (String) confMap.get("fs.defaultFS");
    if (Strings.isNullOrEmpty(fsDir)) {
      fsDir = "file:///";
    }
    config.set("fs.defaultFS", fsDir);
    this.hdfsConfiguration = config;
    RelLocalFileSystemBuilder builder = new RelLocalFileSystemBuilder();
    String dir = System.getProperty("user.dir");
    Map<String, String> map = new HashMap<>();
    map.put(DpRelativeLocalFileSystem.FS_DP_REL_LOCAL_ROOTDIR, dir);
    FileSystem tmpFs = builder.createFileSystem("user", map);
    this.tmpFileSystem = tmpFs;
  }


  @Override
  protected synchronized void checkFileSystem() throws IOException {
    if (fileSystem == null) {
      fileSystem = FileSystem.get(hdfsConfiguration);
    }
  }

  @Override
  public String getPrincipal() {
    return System.getProperty("user.name");
  }


  @Override
  public synchronized DataSource getDataSource(String dsNam, String dbName) throws IOException, SQLException {
    if (dataSource == null) {
      ResourceLoader loader = new DefaultResourceLoader();
      Resource resource = loader.getResource("classpath:/local-debug.properties");
      InputStream inputStream = resource.getInputStream();
      Properties properties = new Properties();
      properties.load(inputStream);
      dataSource = DataSourceUtil.newDataSource(properties);
    }
    return dataSource;
  }

  public Input getConfig() {
    return this.config;
  }

  public FileSystem getHadoopFileSystem() throws IOException {
    checkFileSystem();
    return this.fileSystem;
  }

  @Override
  public FileSystem getFileSystem(String dsName) throws IOException {
    return this.tmpFileSystem;
  }

  @Override
  public FileSystem getTmpFileSystem() {
    return this.tmpFileSystem;
  }

  @Override
  public File getWorkDir() {
    return this.workDir;
  }

  @Override
  public String getTaskName() {
    return this.taskName;
  }

  @Override
  public String getJobName() {
    return this.jobName;
  }

  @Override
  public String getRunTimeJobId() {
    return runTimeJobId;
  }

  @Override
  public String getFlowName() {
    return this.flowName;
  }

  @Override
  public String getFlowExecId() {
    return this.flowExecId;
  }

  @Override
  public String getScheduleId() {
    return null;
  }

  @Override
  public String getConfigId() {
    return this.configId;
  }

  @Override
  public void output(Output output) throws Exception {
    this.pluginService.output(output);
  }


}